package com.atguigu.flink.streamapi.sink;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by Smexy on 2022/11/22
 *
 -- 每种传感器只保留最新的信息
 create table ws(
 id String,
 ts UInt64,
 vc UInt32
 )engine = ReplacingMergeTree(ts)
 order by (id);



 JdbcSink.sink(
     String sql,        //写入的sql
     JdbcStatementBuilder<T> statementBuilder,    //向语句中传参，填充占位符
     JdbcExecutionOptions executionOptions,         //添加执行的参数
     JdbcConnectionOptions connectionOptions       //连接的参数
 );
 */
public class Demo5_JDBCSink
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建JDBCSink
        SinkFunction<WaterSensor> jdbcSink = JdbcSink.sink(
            "insert into ws values(?,?,?)",
            new JdbcStatementBuilder<WaterSensor>()
            {

                //一条数据，调用一次
                @Override
                public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {
                    ps.setString(1, waterSensor.getId());
                    ps.setLong(2, waterSensor.getTs());
                    ps.setInt(3, waterSensor.getVc());

                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchSize(200)  //批量写入，一批的最大条数
                .withMaxRetries(3)   //出现异常，重试的次数    保证数据库支持幂等
                .withBatchIntervalMs(2000) //如果数量上不满足一批，满足时间间隔就写出
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:clickhouse://hadoop102:8123/default")
                .build()
        );


        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .addSink(jdbcSink);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
